/*
 * Copyright 2014 Semidata. All rights reserved.
 */
package com.semidata.trp.report.merge;

import java.io.File;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.TreeMap;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import com.semidata.trp.report.data.UserData;

/**
 * @author shewei@semidata.com, She Wei
 * 
 */
public class MonitorMergeThread {

	private static class DataBlock {
		private static int MAX_SIZE = 1024;
		private List<UserData> queue = new ArrayList<>(MAX_SIZE);
		private int offset = 0;
		private char suffix;

		public DataBlock(char suffix) {
			this.suffix = suffix;
		}

		public boolean add(UserData data) {
			if (queue.size() >= MAX_SIZE)
				return false;
			queue.add(data);
			return true;
		}

		public UserData get() {
			if (offset >= queue.size())
				return null;
			UserData data = queue.get(offset);
			offset++;
			return data;
		}

		public Object getSuffix() {
			return suffix;
		}
	}

	

	private BlockingQueue<DataBlock> queue;
	private File outDir;
	private Map<Character, DataBlock> blocks;
	private Thread thread = null;
	private boolean quit = false;
	private Map<Character, TreeMap<String, UserData>> trees;

	public MonitorMergeThread(List<Character> suffixes, File outDir) {

		this.queue = new LinkedBlockingQueue<DataBlock>();
		this.trees = new HashMap<>();
		this.blocks = new HashMap<>();
		for (char ch : suffixes) {
			blocks.put(ch, new DataBlock(ch));
			trees.put(ch, new TreeMap<String, UserData>());
		}
		this.outDir = outDir;
	}

	private void output(char suffix, TreeMap<String, UserData> map) {
		MergeOutput out = new MergeOutput(outDir, suffix);
		for (UserData data : map.values()) {
			out.write(data.toString());
		}
		out.close();
	}

	private void mergeBlock(DataBlock block) {
		TreeMap<String, UserData> map = trees.get(block.getSuffix());
		UserData data;
		while ((data = block.get()) != null) {
			UserData userData = map.get(data.getUserId());

			if (userData != null) {
				userData.merge(data);
			} else {
				map.put(data.getUserId(), data);
			}
		}
	}

	public void merge() {

		thread = new Thread(new Runnable() {

			@Override
			public void run() {

				while (!quit || !queue.isEmpty()) {
					DataBlock block = null;
					try {
						block = queue.poll(10, TimeUnit.MILLISECONDS);
					} catch (InterruptedException e) {
					}

					if (block == null)
						continue;
					mergeBlock(block);
				}

				for (DataBlock block : blocks.values()) {
					mergeBlock(block);
				}

				for (Entry<Character, TreeMap<String, UserData>> entry : trees
				        .entrySet()) {
					output(entry.getKey(), entry.getValue());
				}
			}
		});

		thread.start();

	}

	public void quit() {
		this.quit = true;
	}

	public void join() {

		try {
			if (thread != null)
				thread.join();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

	public void add(UserData data) {

		char ch = data.getUserId().charAt(0);
		DataBlock block = blocks.get(ch);
		if (block == null)
			return;

		if (!block.add(data)) {
			queue.offer(block);
			block = new DataBlock(ch);
			block.add(data);
			blocks.put(ch, block);
		}
	}

}
